package com.superiron.scaffold.core.service.mq;

import cn.hutool.core.util.RandomUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;

/**
 * @author ：yzg
 * @date ：Created in 2022/8/29 14:07
 * @description：
 * @modified By：
 * @version: $
 */
@Slf4j
public class MqUtil {
	
	private MqUtil() {
	}
	
	public static DefaultMQProducer getProducer(String producerId) throws Exception {
		if (StringUtils.isNotBlank(producerId)) {
			if (MapUtils.isEmpty(InitMqProducer.producerMap) || !InitMqProducer.producerMap.containsKey(producerId)) {
				DefaultMQProducer defaultMQProducer = null;
				try {
					defaultMQProducer = MqUtil.getDoubleCheckSingletonInstance(producerId);
					InitMqProducer.producerMap.put(producerId, defaultMQProducer);
				} catch (MQClientException e) {
					log.error("创建生产者失败：{}", e.getLocalizedMessage());
				}
				return defaultMQProducer;
			} else {
				return InitMqProducer.producerMap.get(producerId);
			}
		} else {
			throw new Exception("没有找到生产者");
		}
	}
	
	/**
	 * 双重加锁验证
	 */
	public static DefaultMQProducer getDoubleCheckSingletonInstance(String namesrvAddr) throws MQClientException {
		DefaultMQProducer defaultMQProducer = null;
		if(InitMqProducer.producerMap.get(namesrvAddr) == null){
			synchronized(DefaultMQProducer.class){
				if (defaultMQProducer == null) {
					defaultMQProducer = createDefaultMqProducer(namesrvAddr);
				}
			}
		}
		return defaultMQProducer;
	}
	
	private static DefaultMQProducer createDefaultMqProducer(String namesrvAddr) throws MQClientException {
		DefaultMQProducer defaultMQProducer = new DefaultMQProducer(RandomUtil.randomString(11));
		defaultMQProducer.setNamesrvAddr(namesrvAddr);
		defaultMQProducer.setInstanceName(RandomUtil.randomString(11));
		//客户端回调线程数，默认：Runtime.getRuntime.availableProcessors，当前cpu核心数
		defaultMQProducer.setClientCallbackExecutorThreads(4);
		//持久化消费者时间间隔
		defaultMQProducer.setPersistConsumerOffsetInterval(5000);
		defaultMQProducer.setSendMsgTimeout(10000);//生产发送消息超时时间
		defaultMQProducer.setCompressMsgBodyOverHowmuch(4096); //消费体容量上限，默认是4m
		defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(false); //是否在内部发送失败时重试另一个broker
		defaultMQProducer.setRetryTimesWhenSendFailed(2);//同步模式下重试次数限制
		defaultMQProducer.setRetryTimesWhenSendAsyncFailed(2); //异步模式下重试次数限制
		defaultMQProducer.start();
		return defaultMQProducer;
	}
}
